package com.atguigu.gulimall.zzmycanal.config;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.InsertListenPoint;
import com.xpand.starter.canal.annotation.ListenPoint;
import com.xpand.starter.canal.annotation.UpdateListenPoint;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@CanalEventListener
public class BusinessListener {
 public static ExecutorService executor = Executors.newFixedThreadPool(10);

 // @InsertListenPoint
 @ListenPoint(schema = "mytestdb", table = {"mytable1"})
 public void adUpdateOrAdd( CanalEntry.RowData rowData)throws  Exception {
  Thread.sleep(200);
  CompletableFuture.runAsync(() -> {
   System.err.println("mytable1______");
  }, executor).thenRunAsync(()->{
   showData(rowData);
  },executor);


 }

 @ListenPoint(schema = "mytestdb", table = {"mytable2"})
 public void adUpdateOrAdd1(CanalEntry.RowData rowData) {
  System.err.println("mytable2______");
  showData(rowData);
 }
 private void showData(CanalEntry.RowData rowData) {
  List<CanalEntry.Column> c1 = rowData.getBeforeColumnsList();
  List<CanalEntry.Column> c2 = rowData.getAfterColumnsList();
  if (c1.size() > 0 && c2.size() > 0) { //表示update
   for (int i = 0; i < c1.size(); i++) {
    CanalEntry.Column cx = c1.get(i);
    CanalEntry.Column cy = c2.get(i);
    if (!cx.getValue().equals(cy.getValue())) {
     System.out.print("更新数据....");
     System.out.println(cx.getName() + ":" + cx.getValue() + "--->" + cy.getName() + ":" + cy.getValue());
    }
   }
  }
  if (c1.size() == 0 && c2.size() > 0) {//表示新增
   System.out.print("添加数据....");
   for (int i = 0; i < c2.size(); i++) {
    CanalEntry.Column cy = c2.get(i);
    if (i == c2.size()) {
     System.out.println(cy.getName() + ":" + cy.getValue() + "   ");
    }
    System.out.print(cy.getName() + ":" + cy.getValue() + "   ");

   }

  }
 }


}